package com.heima.wemedia.scan.listener;

import com.heima.model.wemedia.entity.KafkaMessageConstants;
import com.heima.wemedia.scan.service.WmNewsAutoScanService;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class NewsAutoScanListener {

    @Resource
    private WmNewsAutoScanService wmNewsAutoScanService;

    @KafkaListener(topics = KafkaMessageConstants.WM_NEWS_AUTO_SCAN_TOPIC,groupId = "WM_NEWS_AUTO_SCAN")
    public void newsAutoScanListener(ConsumerRecord<String, String> record){

        String value = record.value();

        if(StringUtils.isNotBlank(value)){

            wmNewsAutoScanService.autoScanNews(Long.valueOf(record.value()));

        }

    }
}
